package org.dbstar.scheduler.job;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.dbstar.scheduler.beans.CompoundJob;
import org.dbstar.scheduler.beans.CompoundJobResult;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.beans.JobConf;
import org.dbstar.scheduler.beans.JobDefine;
import org.dbstar.scheduler.beans.LoopJobConf;
import org.dbstar.scheduler.control.JobController;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.identifier.IdentifierFactory;
import org.dbstar.scheduler.identifier.IdentifierType;
import org.dbstar.scheduler.utils.JobUtils;


/**
 * 构建使用LoopJobConf的CompoundJob
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-5-7
 */
public class LoopJobFactory extends AbstractJobFactory<CompoundJob> {
	private final JobController jobController;
	private final IdentifierFactory identifierFactory;

	public LoopJobFactory(JobController jobController, IdentifierFactory identifierFactory) {
		super();
		this.jobController = jobController;
		this.identifierFactory = identifierFactory;
	}

	@Override
	public CompoundJob createJob(SchedulerFactory schedulerFactory, JobConf conf, String submitter,
			Map<String, String> parameters, long submitTime) throws IllegalFactoryException {
		if (conf == null) {
			throw new IllegalFactoryException("Job Conf is null");
		}
		if (StringUtils.isBlank(submitter)) {
			throw new IllegalFactoryException("Job Submitter is null");
		}
		CompoundJob job = new CompoundJob();
		job.setJobConfIdentifier(conf.getIdentifier());
		job.setSubmitTime(submitTime);
		job.setSubmitter(submitter);
		job.setParameters(JobUtils.getJobParameters(conf, parameters, true));
		try {
			build(job, (LoopJobConf) conf, submitter, schedulerFactory);
		} catch (DaoException e) {
			throw new IllegalFactoryException(e);
		}
		return job;
	}

	private void build(final CompoundJob job, final LoopJobConf conf, final String submitter,
			final SchedulerFactory schedulerFactory) throws IllegalFactoryException, DaoException {
		final List<JobDefine> list = new ArrayList<JobDefine>();
		final CompoundJobResult result = (CompoundJobResult) job.getResult();
		final JobDefine jd = conf.getSubJob();

		loop(job, conf, new LoopCallback() {
			@Override
			public void loopAt(int index, String loopVarName, String loopVarValue) throws IllegalFactoryException,
					DaoException {
				Map<String, String> parameters = new HashMap<String, String>();
				if (job.getParameters() != null) {
					parameters.putAll(job.getParameters());
				}
				parameters.put(loopVarName, loopVarValue);

				Set<Integer> dependencies = null;
				if (conf.isLoopDependency() && index > 0) {
					dependencies = new HashSet<Integer>();
					dependencies.add(index);
				}

				boolean skip = !JobUtils.matchCondition(jd.getCondition(), parameters);
				if (skip) {
					Identifier identifier = identifierFactory.createIdentifier(IdentifierType.skip);
					list.add(new JobDefine(identifier, JobUtils.formatSubJobParameters(jd.getParameters(), parameters),
							dependencies, jd.getCondition(), skip));
					result.addSubJobResult(identifier, JobStatus.init, false, dependencies);
				} else {
					Job subJob = schedulerFactory.createJob(jd.getIdentifier(),
							JobUtils.formatSubJobParameters(jd.getParameters(), parameters), submitter);
					list.add(new JobDefine(jobController.addJob(subJob), subJob.getParameters(), dependencies, jd
							.getCondition(), skip));
					result.addSubJobResult(subJob, dependencies);
				}
			}
		});

		job.setSubJobs(list);
	}

	private void loop(CompoundJob job, LoopJobConf conf, LoopCallback callback) throws IllegalFactoryException,
			DaoException {
		int loopStep = conf.getLoopStep();
		if (loopStep == 0) {
			throw new IllegalFactoryException("循环步长不能为0");
		}
		String loopVar = conf.getLoopVar();
		if (StringUtils.isBlank(loopVar)) {
			throw new IllegalFactoryException("循环变量不能为空");
		}

		String loopType = conf.getLoopType();
		String loopFrom = JobUtils.formatByParameter(conf.getLoopFrom(), job.getParameters());
		String loopTo = JobUtils.formatByParameter(conf.getLoopTo(), job.getParameters());
		if ("int".equals(loopType)) {
			int from;
			try {
				from = Integer.parseInt(loopFrom);
			} catch (NumberFormatException e) {
				throw new IllegalFactoryException("循环开始值不正确：" + loopFrom, e);
			}
			int to;
			try {
				to = Integer.parseInt(loopTo);
			} catch (NumberFormatException e) {
				throw new IllegalFactoryException("循环结束值不正确：" + loopTo, e);
			}
			for (int index = 0; loopStep > 0 ? from <= to : from >= to; from += loopStep) {
				callback.loopAt(index++, loopVar, Integer.toString(from));
			}
		} else {
			DateFormat format;
			int field;
			if ("hour".equals(loopType)) {
				format = new SimpleDateFormat("yyyyMMdd HH");
				field = Calendar.HOUR_OF_DAY;
			} else if ("day".equals(loopType)) {
				format = new SimpleDateFormat("yyyyMMdd");
				field = Calendar.DAY_OF_MONTH;
			} else if ("week".equals(loopType)) {
				format = new SimpleDateFormat("yyyyMMdd");
				field = Calendar.WEEK_OF_YEAR;
			} else if ("month".equals(loopType)) {
				format = new SimpleDateFormat("yyyyMMdd");
				field = Calendar.MONTH;
			} else {
				throw new IllegalFactoryException("未知的循环类型：" + loopType);
			}
			Calendar from = Calendar.getInstance();
			try {
				from.setTime(format.parse(loopFrom));
			} catch (ParseException e) {
				throw new IllegalFactoryException("循环开始值不正确：" + loopFrom, e);
			}
			Date to;
			try {
				to = format.parse(loopTo);
			} catch (ParseException e) {
				throw new IllegalFactoryException("循环结束值不正确：" + loopTo, e);
			}
			for (int index = 0; loopStep > 0 ? !from.getTime().after(to) : !from.getTime().before(to);) {
				callback.loopAt(index++, loopVar, format.format(from.getTime()));
				from.add(field, loopStep);
			}
		}
	}

	private interface LoopCallback {
		public void loopAt(int index, String loopVarName, String loopVarValue) throws IllegalFactoryException,
				DaoException;
	}
}
